package commons.base.task;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * 批处理任务执行器。包含一个队列和一个线程
 * @author yuan<cihang.yuan@happyelements.com>
 *
 */
public abstract class BatchExecutor<E> extends Thread {

	private static final Logger log = LogManager.getLogger(BatchExecutor.class);
	
	/** 批大小 */
	private int batchSize;
	/** 超时时间, 单位秒 */
	private int timeout;
	
	private final Object lock = new Object();
	private BlockingQueue<E> queue;
	private AtomicBoolean stopping = new AtomicBoolean(false);
	
	public BatchExecutor(String name, int batchSize, int maxQueueSize, int timeout){
		super(name);
		
		this.batchSize = batchSize;
		this.timeout = timeout;
		
		queue = new LinkedBlockingQueue<E>(maxQueueSize);
	}
	
	public void addBatch(E e){
		if (stopping.get()) {
			return;
		}
		
		boolean success = queue.offer(e);
		if(success && (queue.size() >= batchSize)){
			signal();
		}
	}
	
	@Override
	public void run(){
		while (!stopping.get()) {
			try {
				await();
			} catch (InterruptedException e) {
				log.error(e.getMessage(), e);
			}
			
			if(queue.size() >= batchSize){
				while((queue.size() >= batchSize)){
					processQueue(batchSize);
				}
			}else{
				processQueue(queue.size());
			}
		}
	}
	
	private void processQueue(int size) {
		if (size <= 0) {
			return ;
		}
		
		List<E> list = new ArrayList<E>();
		
		for (int i=0; i<size; i++) {
			E e = queue.poll();
			if (e != null) {
				list.add(e);
			}
		}
		
		try {
			submit(list);
		} catch (Throwable e) {
			log.error(e.getMessage(), e);
		}
	}
	
	/**
	 * 处理一批数据
	 * @param list
	 * @throws Throwable
	 */
	protected abstract void submit(List<E> list) throws Throwable;
	
	private void processAll(){
		while(queue.size() > 0){
			int size = queue.size() > batchSize ? batchSize : queue.size();
			processQueue(size);
		}
	}

	public void shutdown() throws InterruptedException{
		stopping.set(true);
		signal();
		
		join();
		
		processAll();
	}
	
	private void await() throws InterruptedException{
		synchronized (lock) {
			lock.wait(timeout * 1000);
		}
	}
	
	private void signal(){
		synchronized (lock) {
			lock.notifyAll();
		}
	}
}
